package com.atbeijing.D08;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * flink将数据写到mysql
 *
 * 一般不用,因为mysql吞吐量小,容易被冲爆
 */
public class Example2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .addSink(new RichSinkFunction<SensorReading>() {
                    private Connection conn;
                    private PreparedStatement insertStmt;
                    private PreparedStatement updateStmt;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        conn = DriverManager.getConnection(
                                "jdbc:mysql://localhost:3306/mytest",
                                "root",
                                "123456"
                        );
                        insertStmt = conn.prepareStatement("INSERT INTO temps (id, temp) VALUES (?, ?)");
                        updateStmt = conn.prepareStatement("UPDATE temps SET temp = ? WHERE id = ?");
                    }

                    @Override
                    public void invoke(SensorReading value, SinkFunction.Context context) throws Exception {
                        updateStmt.setDouble(1, value.temperature);
                        updateStmt.setString(2, value.id);
                        updateStmt.execute();

                        //如果更新失败就表示数据不存在,就插入数据
                        if (updateStmt.getUpdateCount() == 0) {
                            insertStmt.setString(1, value.id);
                            insertStmt.setDouble(2, value.temperature);
                            insertStmt.execute();
                        }
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        insertStmt.close();
                        updateStmt.close();
                        conn.close();
                    }
                });

        env.execute();
    }
}
